package com.Reactor3参考文档;

import java.util.ArrayList;
import java.util.List;

import org.junit.Test;

import reactor.core.publisher.Flux;

/**推送（push）模式

create 的一个变体是 push，适合生成事件流。与 create`类似，`push 也可以是异步地， 并且能够使用以上各种溢出策略（overflow strategies）管理背压。每次只有一个生成线程可以调用 next，complete 或 error。
 * @author DELL
 *
 */
public class Push {

	// 假设你有一个监听器 API，它按 chunk 处理数据，有两种事件：（1）一个 chunk 数据准备好的事件；（2）处理结束的事件
	public static interface MyEventListener<T> {
	    void onDataChunk(List<T> chunk);
	    void processComplete();
	    void processError(Throwable e);
	}

	public static class MyEventProcessor {

		private MyEventListener<String> myEventListener;

		public void register(MyEventListener<String> myEventListener) {
			this.myEventListener = myEventListener;
		}

		public void add(ArrayList<String> chunk) {
			myEventListener.onDataChunk(chunk);
		}

	}

	@Test
	public void test() {
		MyEventProcessor myEventProcessor = new MyEventProcessor();

		// 所有这些都是在 myEventProcessor 执行时异步执行的。
		Flux<String> bridge = Flux.push(sink -> {
			// 桥接 MyEventListener
			myEventProcessor.register(new MyEventListener<String>() {
				public void onDataChunk(List<String> chunk) {
					for(String s : chunk) {
						sink.next(s);
					}
				}

				public void processComplete() {
					sink.complete();
				}

				public void processError(Throwable e) {
					sink.error(e);
				}
			});
		});
		bridge.subscribe(System.out::println);

		myEventProcessor.add(new ArrayList<String>(){{add("aaa");add("bbb");add("ccc");}});
	}

}
